package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.OrderCoreService;
import com.mall.promo.dto.SecKillRequest;
import org.apache.dubbo.config.annotation.Reference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * @author
 * @date 2020/12/27 14:20
 */
@Component
public class PromoOrderConsumer {
    private DefaultMQPushConsumer mqPushConsumer;
    public static final String PROMO_ORDER_SECKILL_TOPIC = "promo_order_seckill";

    @Reference(check = false)
    OrderCoreService orderCoreService;

    @PostConstruct
    public void init() {
        mqPushConsumer = new DefaultMQPushConsumer("promo_order_consumer");
        mqPushConsumer.setNamesrvAddr("127.0.0.1:9876");
        try {
            mqPushConsumer.subscribe(PROMO_ORDER_SECKILL_TOPIC, "*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        mqPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msg, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try {
                    MessageExt messageExt = msg.get(0);
                    byte[] body = messageExt.getBody();
                    String bodyStr = new String(body);
                    SecKillRequest request = JSON.parseObject(bodyStr, SecKillRequest.class);

                    //创建订单 创建物流信息
                    orderCoreService.createPromoOrder(request);

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    e.printStackTrace();

                }
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        try {
            mqPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
